package com.ft.turorial.spring.boot.kafka.consumer;

import com.alibaba.fastjson.JSON;
import com.ft.turorial.spring.boot.kafka.message.MessageWrapper;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;

import java.util.Optional;
import java.util.concurrent.CountDownLatch;

/**
 * \* Created with IntelliJ IDEA.
 * \* User: Administrator
 * \* Date: 2016/11/30
 * \* Time: 11:42
 * \* To change this template use File | Settings | File Templates.
 * \* Description:
 * \
 */
public class Receiver {
    private static final Logger LOGGER = LoggerFactory
            .getLogger(Receiver.class);

    private CountDownLatch latch = new CountDownLatch(1);

    @KafkaListener(topics = "helloworld.t")
    public void receiveMessage(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            try {
                //如果是JSON格式字符串，开使用 JSON.parseObject来转换成具体对象，例如自定义的消息对象
                MessageWrapper messageWrapper = JSON.parseObject(message.toString(), MessageWrapper.class);

                LOGGER.info("received the type of message is'{}', message='{}'",
                        messageWrapper.getMessageType(),messageWrapper.getData().toString());
            } catch (Exception e) {
                LOGGER.error("将接收到的消息进行处理时异常, 消息:{}, 异常:{}",message.toString(),e);
            }
        }
        latch.countDown();
    }

    public CountDownLatch getLatch() {
        return latch;
    }
}
